# coding:utf-8
"""
sqoop 将 mysql 库中的文件导入到 hdfs中
mysql 中表结构
+-----------------+--------------+------+-----+-------------------+----------------+
| Field           | Type         | Null | Key | Default           | Extra          |
+-----------------+--------------+------+-----+-------------------+----------------+
| id              | bigint(20)   | NO   | PRI | NULL              | auto_increment |
| action          | varchar(40)  | NO   | MUL | NULL              |                |
| imei            | varchar(50)  | NO   |     | NULL              |                |
| os              | varchar(15)  | NO   |     | NULL              |                |
| version_no      | varchar(10)  | NO   |     | NULL              |                |
| sign            | varchar(200) | NO   |     | NULL              |                |
| token_id        | varchar(50)  | YES  |     | NULL              |                |
| req_encrypt_msg | longtext     | YES  |     | NULL              |                |
| req_data        | longtext     | YES  |     | NULL              |                |
| res_data        | longtext     | YES  |     | NULL              |                |
| res_encrypt_msg | longtext     | YES  |     | NULL              |                |
| create_time     | timestamp    | NO   |     | CURRENT_TIMESTAMP |                |
+-----------------+--------------+------+-----+-------------------+----------------+
+-------------------+
| Tables_in_ibd_app |
+-------------------+
| app_log_201611    |
| app_log_201612    |
| app_log_201701    |
| app_log_201702    |
| app_log_201703    |
| app_log_201704    |
| app_log_201705    |
| app_log_201706    |
| app_log_201707    |
| app_log_201708    |
| app_log_201709    |
| app_log_201710    |
| app_log_201711    |
| app_log_201712    |
| app_log_201801    |
| app_log_201802    |
| app_log_201803    |
| app_log_201804    |
| app_log_201805    |
| app_log_201806    |
| app_log_201807    |
| app_log_201808    |
+-------------------+
每个表结构都一样，按月记录 api 访问量
"""
def mysql_conn(pymysql,**kwargs):
    """
    :param pymysql:  pymysql module
    :param kwargs dict key = value
		host="hostname"
		user="dbuser"
		password="password"
		port = int("port")
		dbname = "dbname"
		charset = "charset"
    :return: pymysql.cursors.Cursor
    """

    conn = pymysql.connect(host=kwargs["host"],user=kwargs["user"],passwd=kwargs["password"],
                           db=kwargs["dbname"],port=kwargs["port"],charset=kwargs["charset"])

    cursor = conn.cursor()
    return  cursor

def get_db_all_table(pymysql,**kwargs):

    """
    :param pymysql:  pymysql module
    :param kwargs dict key = value
		host="hostname"
		user="dbuser"
		password="password"
		port = int("port")
		dbname = "dbname"
		charset = "charset"
    :return: table_list = [tablename1,tablename2,tablename3,...,tablenamen]
    """

    cursor = mysql_conn(pymysql,**kwargs)
    sql = "show tables from" + kwargs["dbname"]
    cursor.execute("show tables from ibd_app")
    result = cursor.fetchall()
    table_list = []
    for tablename in result:
        table_list.append(tablename[0])
    return table_list

def get_table_line_num(pymysql,**kwargs):

    cursor = mysql_conn(pymysql,**kwargs)
    table_list = get_db_all_table(pymysql,**kwargs)
    table_line_list = []
    sql = "SELECT COUNT(*) FROM "
    for tablename in table_list:
        cursor.execute(sql + tablename)
        count = cursor.fetchall()
        table_line_list.append(count[0][0])
    
    return table_line_list

def main():
    #kw = {"host":"master","user":"root","password":"123456","port":3306,"dbname":"ibd_app","charset":"utf8"}
    #import pymysql
    #table_line_list = get_table_line_num(pymysql,**kw)
    #print(table_line_list)
    sqoop_cmd()
def connect_hdfs(url,dirname):
	import hdfs
	client = hdfs.Client(url)
	try:
		client.status(dirname)
		return 0
	except BaseException :
		return 1

def sqoop_cmd():
    import os
    kw = {"host":"master","user":"root","password":"123456","port":3306,"dbname":"ibd_app","charset":"utf8"}
    import pymysql
    table_list = get_db_all_table(pymysql,**kw)
    table_line_list = get_table_line_num(pymysql,**kw)
    sql = 'SELECT TABLE_NAME FROM information_schema.tables WHERE table_schema = '
    prefer = "'"
    select = "select id, action, imei, os, version_no, req_data, res_data, create_time from "
    where = " where $CONDITIONS LIMIT "
    last = "'"
    sqoop_connect = "sqoop import --connect jdbc:mysql://master:3306/ibd_app --username root --password 123456 "
    sqoop_file_format = "--as-parquetfile "
    sqoop_map_num = "-m 1 "
    sqoop_fields_terminated = "--fields-terminated-by '\\t'"
    for i in range(len(table_list)):
        sqoop_query = "--query " + prefer + select + table_list[i] + where + str(table_line_list[i]) + last + " "
        // 每个表导入 hdfs 目录不一样，如果这里不使用 --query 这个参数，会导致，导入 hdfs 上的中文内容是乱码，不知道啥原因
        // 本来这里使用 hdfs python api 检测目标目录是否存在，如果存在就删除的，可是发现了很奇怪的问题，dirpath 变量写死是没有问题的，
        // 通过字符串拼接就有问题，总是说找不到目录，可是目录确实存在。写死和字符串拼接打印出来一模一样，不知道为什么路径不能变化，非要写死
        sqoop_tagdir = "--target-dir /sqoop/ibd_app/" + table_list[i] + " "
        // 拼接 sqoop 命令，本来想用shell 脚本拼接的，可是 shell 脚本拼接的时候总是出问题，主要表现为： --query 'select',id
        sqoop = sqoop_connect + sqoop_query + sqoop_file_format + sqoop_map_num +sqoop_tagdir + sqoop_fields_terminated
        print(sqoop)

        os.system(sqoop)

if __name__ == '__main__':
	main()

"""
这个函数检测hdfs 上的某个目录是否存在，dirpath 这个变量写死就可以准确的获取到目录，
def connect_hdfs(url,dirpath):
	import hdfs
	client = hdfs.Client(url)
	try:
        # 如果目录不存在会抛异常 HdfsError 这个异常
		client.status(dirpath)
		return 0
	except BaseException :
    	return 1
url="http://master:50070"   注意这个url 很古怪，不是namenode 的 9000端口，而是 hdfs web ui 的端口
index = sqoop_tagdir.index("/")
dirpath = sqoop_tagdir[index :]

connect_hdfs(url,dirpath)
return 0
这里测试是成功的，只是 "sqoop_tagdir" 这个变量的 tablename 部分是手动指定的，不是通过连接 mysql 自动获取的，现在怀疑问题就出在这里，有待测试
 
"""